package mq

import (
	"fmt"
	"reflect"
	"strings"

	"yingsheng.com/go-api-server/errors"
	"yingsheng.com/go-api-server/log"
)

var (
	mqs       mqType
	producers = make(map[string]reflect.Type)
	consumers = make(map[string]reflect.Type)
)

type mqType struct {
	list        []MQInterface
	recoverFunc func(err errors.RecoverErrorInterface)
}

func (m *mqType) SetRecoverConsumerPanic(f func(err errors.RecoverErrorInterface)) {
	m.recoverFunc = f
}

func (m *mqType) RecoverConsumerPanic(err errors.RecoverErrorInterface) {
	if e := recover(); e != nil {
		errMsg := fmt.Sprintf("%v", e)
		err.SetError(errMsg)

		if m.recoverFunc != nil {
			//执行自定义Recover事件
			m.recoverFunc(err)
		} else {
			//默认Recover事件
			log.Logger.Critical("mq panic：", errMsg)
		}
	}
}

type MQInterface interface {
	Stop() error
}

type ProducerInterface interface {
	Init(topic string) error
	Publish(msg []byte) error
	Stop() error
}

type ConsumerInterface interface {
	Init(topic, channel string) error
	Handler(chi ConsumerHandlerInterface) error
	Stop() error
}

type ConsumerHandlerInterface interface {
	HandlerMessage(msg []byte) error
}

func SetRecoverConsumerPanic(f func(err errors.RecoverErrorInterface)) {
	mqs.SetRecoverConsumerPanic(f)
}

func Exit() error {
	var (
		err    error
		errMsg string
	)
	for _, v := range mqs.list {
		err = v.Stop()
		if err != nil {
			if len(errMsg) > 0 {
				errMsg += ","
			}
			errMsg += fmt.Sprintf("%q", err)
		}

		if len(errMsg) > 0 {
			err = fmt.Errorf("MQ: Exit error %s", errMsg)
		}
	}

	return err
}

func NewProducer(adapter, topic string) (pi ProducerInterface, err error) {
	k := strings.ToLower(adapter)

	v, ok := producers[k]
	if !ok {
		err = fmt.Errorf("MQ: unknown Producer adapter %q", adapter)
		return
	}

	vo := reflect.New(v)
	pi, ok = vo.Interface().(ProducerInterface)
	if !ok {
		err = fmt.Errorf("MQ: %q is a invalid ProducerInterface", adapter)
		return
	}

	err = pi.Init(topic)
	if err == nil {
		mqs.list = append(mqs.list, pi)
	}

	return
}

func NewConsumer(adapter, topic, channel string, chi ConsumerHandlerInterface) (ci ConsumerInterface, err error) {
	k := strings.ToLower(adapter)

	v, ok := consumers[k]
	if !ok {
		err = fmt.Errorf("MQ: unknown Consumer adapter %q", adapter)
		return
	}

	vo := reflect.New(v)
	ci, ok = vo.Interface().(ConsumerInterface)
	if !ok {
		err = fmt.Errorf("MQ: %q is a invalid ConsumerInterface", adapter)
		return
	}

	err = ci.Init(topic, channel)
	if err != nil {
		return
	}

	err = ci.Handler(chi)
	if err != nil {
		return
	}

	mqs.list = append(mqs.list, ci)

	return
}
